package com.zhike.function;

import com.alibaba.druid.util.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhike.bean.TableProcess;
import com.zhike.constant.HbaseConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * @Author: zhike
 * @CreateTime: 2024/1/18
 * @Description:  广播处理函数
 */
@Slf4j
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {

    private Connection connection;
    private PreparedStatement preparedStatement;

    private MapStateDescriptor<String,TableProcess> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String,TableProcess> mapStateDescriptor){
        this.mapStateDescriptor = mapStateDescriptor;
    }
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(HbaseConstant.PHOENIX_SERVER);
        System.out.println("获取连接成功");
    }

    //处理主流
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //1.获取广播流的配置数据(广播状态)
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //获取主流数据的来源表
        String table = value.getJSONObject("source").getString("table");
        TableProcess tableProcess = broadcastState.get(table);
        if (tableProcess != null) {
            JSONObject after = value.getJSONObject("after");
            JSONObject before = value.getJSONObject("before");
            if (before!=null){
                value.put("sinkTable", tableProcess.getSinkTable());
                out.collect(before);
            }
            if (after != null) {
                //2.过滤需要的字段
                filterColumn(after, tableProcess.getSinkColumns());

                //3.添加sinkTable并写出
                value.put("sinkTable", tableProcess.getSinkTable());
                out.collect(value);
            }

        }else {
            log.info("找不到对应的维度表:" + table);
        }
    }

    private void filterColumn(JSONObject after, String sinkColumns) {
        String[] split = sinkColumns.split(",");
        List<String> needColumns = Arrays.asList(split);
        Set<Map.Entry<String, Object>> allData = after.entrySet();
        Iterator<Map.Entry<String, Object>> iterator = allData.iterator();
        while (iterator.hasNext()){
            Map.Entry<String, Object> column = iterator.next();
            if (!needColumns.contains(column.getKey())){
                iterator.remove();
            }
        }

    }


    //处理广播流
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        //1.获取并解析数据
        JSONObject jsonObject = JSON.parseObject(value);
        TableProcess tableProcess = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);

        System.out.println(tableProcess.toString());

        //2.校验并建表
        checkTable(tableProcess.getSinkTable(),
                tableProcess.getSinkColumns(),
                tableProcess.getSinkPk(),
                tableProcess.getSinkExtend());

        //3.保存状态，并广播出去
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(tableProcess.getSourceTable(),tableProcess);


    }

    //建表
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        try {
            //1.处理特殊数据
            if (StringUtils.isEmpty(sinkPk)) {
                //如果没有主键默认设置为id
                sinkPk="id";
            }
            if (sinkExtend == null){
                sinkExtend="";
            }

            //2.拼接sql create table if not exists db.table(id varchar primary key,name varchar) xxx
            StringBuilder createTableSql = new StringBuilder("create table if not exists ")
                    .append(HbaseConstant.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");

            String[] columns = sinkColumns.split(",");
            String[] pks = sinkPk.split(",");
            List<String> list = Arrays.asList(pks);
            for (int i = 0; i < columns.length; i++) {
                if (list.contains(columns[i])) {
                    createTableSql.append(columns[i])
                            .append(" varchar not null");
                } else {
                    createTableSql.append(columns[i])
                            .append(" varchar");
                }
                if (i < columns.length - 1) {
                    createTableSql.append(",");
                }
            }
            createTableSql.append(" constraint pk primary key (")
                    .append(sinkPk)
                    .append("))")
                    .append(sinkExtend);

            //3.编译sql
            preparedStatement = connection.prepareStatement(createTableSql.toString());

            //4.执行sql
            preparedStatement.execute();
        } catch (SQLException e) {
            throw new RuntimeException("建表失败："+sinkTable);
        } finally {
            //关闭连接
            if (preparedStatement !=null) {
                try {
                    preparedStatement.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }


    }


}
